/*============================================================================
  File:      pullcaseprocessor.cpp
 
  Summary:   Implementation of PULLCASEPROCESSOR class that is used to extract
			 data from Analysis Services in a 'pull' manner
 
  Date:		 June 30, 2004
------------------------------------------------------------------------------
  This file is part of the Microsoft SQL Server Code Samples.
 
  Copyright (C) 2003 Microsoft Corporation.  All rights reserved.
 
This source code is intended only as a supplement to Microsoft
Development Tools and/or on-line documentation.  See these other
materials for detailed information regarding Microsoft code samples.
 
THIS CODE AND INFORMATION ARE PROVIDED "AS IS" WITHOUT WARRANTY OF ANY
KIND, EITHER EXPRESSED OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE
IMPLIED WARRANTIES OF MERCHANTABILITY AND/OR FITNESS FOR A
PARTICULAR PURPOSE.
============================================================================*/

#include "stdafx.h"


const		DWORD		PULL_CASESET_CACHE_SIZE =	256*1024;		// default size for the cache of the pull case processor, 256K

#include "pullcaseprocessor.h"
#include "DataValues.h"



// General creation method
HRESULT	PullCASE::CreatePullCASE(IDMMemoryAllocator*	pmalloc, PullCASE**	out_ppCase)
{
	ATLASSERT( pmalloc );
	ATLASSERT( out_ppCase );

	HRESULT hRet	=	pmalloc->Alloc( sizeof(PullCASE), (BYTE**)out_ppCase);
	RETURN_ON_FAIL(hRet);

	memset( *out_ppCase, 0, sizeof(PullCASE) );
	return S_OK;
}
// General deletion method
HRESULT	PullCASE::DeletePullCASE(IDMMemoryAllocator*	pmalloc, PullCASE*	in_pCase)
{
	ATLASSERT( pmalloc );
	ATLASSERT( in_pCase );

	
	if( in_pCase->rgCase  != NULL )
	{
		pmalloc->Free((BYTE*)in_pCase->rgCase );
	}
	pmalloc->Free((BYTE*)in_pCase );

	return S_OK;
}


PullCASEQueue::PullCASEQueue()
{
	_pFirst					=	NULL;
	_pLast					=	NULL;
	_spidmmemoryallocator	=	NULL;
	_dwTotalSize			=	0;
	_dwMaxSize				=	PULL_CASESET_CACHE_SIZE;
}

PullCASEQueue::~PullCASEQueue()
{
	Reset();
	ATLASSERT( _pFirst == NULL );
	ATLASSERT( _pLast == NULL );
	ATLASSERT( _spidmmemoryallocator == NULL );
}

HRESULT PullCASEQueue::Initialize( IDMMemoryAllocator* spidmmmemoryallocator, DWORD dwMaxSize)
{
	if( !spidmmmemoryallocator )
	{
		return E_POINTER;
	}
	_spidmmemoryallocator	=	spidmmmemoryallocator;
	_dwMaxSize				=	dwMaxSize;
	_dwTotalSize			=	0;
	

	// Create an event that is signaled while the queue has data
	// - NULL (default) Use the ACL from the primary or impersonation token of the creator
	// - ManualReset=true (the event will be signaled or not depending on the state of the queue)
	// - initial state = false (not signaled, as the queue is empty)
	// - no name for the event
	_hEvtQueueHasData		=	::CreateEvent( NULL, TRUE, FALSE, NULL);
	if( _hEvtQueueHasData == NULL )
	{
		// If the function fails, the return value is NULL
		// Use GetLastError
		return E_UNEXPECTED;
	}

	// Create an event that is signaled while the queue has room for more data
	// - NULL (default) Use the ACL from the primary or impersonation token of the creator
	// - ManualReset=true (the event will be signaled or not depending on the state of the queue)
	// - initial state = true (signaled, as the queue is empty and it has room for more data)
	// - no name for the event
	_hEvtQueueHasRoom		=	::CreateEvent( NULL, TRUE, TRUE, NULL);
	if( _hEvtQueueHasRoom == NULL )
	{
		// If the function fails, the return value is NULL
		// Use GetLastError
		return E_UNEXPECTED;
	}

	return S_OK;
}

HRESULT PullCASEQueue::Reset()
{
	// Delete all the cases from the cache
	PullCASE*	pCurrent = _pFirst;
	while( pCurrent )
	{
		PullCASE*	pTmp	=	pCurrent->pNext;
		PullCASE::DeletePullCASE( _spidmmemoryallocator, pCurrent);
		pCurrent = pTmp;
	}
	_pFirst			=	NULL;
	_pLast			=	NULL;
	_dwTotalSize	=	0;
	_spidmmemoryallocator	=	NULL;

	// Reset the events
	if( _hEvtQueueHasData  != NULL )
	{
		::CloseHandle( _hEvtQueueHasData );
		_hEvtQueueHasData = NULL;
	}
	if( _hEvtQueueHasRoom  != NULL )
	{
		::CloseHandle( _hEvtQueueHasRoom );
		_hEvtQueueHasRoom = NULL;
	}

	return S_OK;
}

HRESULT		PullCASEQueue::AddCase(PullCASE*	in_pCase)
{
	HRESULT	hAddResult		=	S_OK;

	// Check the state of the queue: wait for the QueueRoom event for 0 ms
	DWORD dwQueueRoomState	=	::WaitForSingleObject( _hEvtQueueHasRoom, 0 );
	bool	fQueueHasRoom	=	(dwQueueRoomState == WAIT_OBJECT_0) ? true : false;
	
	
	if( _pLast == NULL )
	{
		
		ATLASSERT( _pFirst == NULL );
		ATLASSERT( _dwTotalSize == 0 );
		_pFirst	=	in_pCase;
		_pFirst->pNext = NULL;
		_pLast	=	_pFirst;
		_dwTotalSize = in_pCase->dwSize;
		hAddResult	=	S_OK;
	}
	else
	{
		ATLASSERT( _pFirst != NULL );
		ATLASSERT( _pLast->pNext == NULL );
		if( fQueueHasRoom )
		{
			_dwTotalSize += in_pCase->dwSize;
			in_pCase->pNext = NULL;
			_pLast->pNext = in_pCase;
			_pLast	=	in_pCase;
			hAddResult	=	S_OK;
		}
		else
		{
			// Not enough room in the cache -- SHOULD NEVER BE HERE!!!
			ATLASSERT(FALSE);
			hAddResult	=	S_FALSE;
		}
	}

	// If the size of the queue exceeded the maximum size, unsignal the QueueHasRoom event
	if( _dwTotalSize > _dwMaxSize )
	{
		if( !::ResetEvent( _hEvtQueueHasRoom ) )
		{
			hAddResult	=	E_UNEXPECTED;
		}
	}

	// If there is at least one case in the queue, set the QueueHasData event
	if( _pFirst != NULL )
	{
		if( !::SetEvent( _hEvtQueueHasData ) )
		{
			hAddResult	=	E_UNEXPECTED;
		}
	}


	return hAddResult;
}

HRESULT		PullCASEQueue::RemoveCase(PullCASE*&	out_rpCase)
{
	HRESULT	hPopResult		=	S_OK;


	out_rpCase		=	NULL;
	// Check the state of the queue: wait for the QueueRoom event for 0 ms
	DWORD dwQueueDataState	=	::WaitForSingleObject( _hEvtQueueHasData, 0 );
	bool	fQueueHasData	=	(dwQueueDataState == WAIT_OBJECT_0) ? true : false;


	// returns the top case and substracts the size from the total size
	if( !fQueueHasData )
	{
		// No data in the cache -- SHOULD NEVER BE HERE!!!
		ATLASSERT(FALSE);
		// Consistency checks
		ATLASSERT( _pFirst == NULL  );
		ATLASSERT( _dwTotalSize == 0 );
		ATLASSERT( _pLast == NULL );
		out_rpCase	=	NULL;
		hPopResult	=	S_FALSE;
	}
	else
	{
		ATLASSERT( _pLast != NULL );
		ATLASSERT( _dwTotalSize >= _pFirst->dwSize ); 
		out_rpCase = _pFirst;
		_dwTotalSize -= _pFirst->dwSize;
		
		if( _pFirst == _pLast )
		{
			_pFirst	=	NULL;
			_pLast	=	NULL;
		}
		else
		{
			ATLASSERT( _pFirst->pNext != NULL );
			_pFirst	= _pFirst->pNext;	
		}
		hPopResult	=	S_OK;
	}
	
	// If there isn't any case in the queue, reset the QueueHasData event
	if( _pFirst == NULL )
	{
		if( !::ResetEvent( _hEvtQueueHasData ) )
		{
			hPopResult	=	E_UNEXPECTED;
		}
	}

	// If a case was fetched successfully, signal the QueueHasRoom event
	if( out_rpCase != NULL && (hPopResult == S_OK) )
	{
		if( !::SetEvent( _hEvtQueueHasRoom ) )
		{
			hPopResult	=	E_UNEXPECTED;
		}
	}

	return hPopResult;
}

PushToPullCaseProcessor::PushToPullCaseProcessor()
{
	_bTerminated		=	false;
	_hTerminationCode	=	S_OK;
}


HRESULT PushToPullCaseProcessor::Initialize(IDMPushCaseSet*	pidmpushcaseset, DWORD dwMaximumLimit) 
{
	HRESULT hr	=	S_OK;

	ATLASSERT(pidmpushcaseset != NULL);
	_bDensePushSet		= (pidmpushcaseset->IsDense() == S_OK) ? true : false;

	hr	=	_lock.Init();
	RETURN_ON_FAIL(hr);

	hr	= _queueCache.Initialize( _spidmmemoryallocator, dwMaximumLimit );
	RETURN_ON_FAIL(hr);

	_currentCase = NULL;

	// Create an event that is signaled when a termination contidion occurred
	// - NULL (default) Use the ACL from the primary or impersonation token of the creator
	// - ManualReset=true (the event will be signaled or not depending on the state of the processing job)
	// - initial state = false (not signaled, as the processing is not terminated)
	// - no name for the event
	_hTermEvent		=	::CreateEvent( NULL, TRUE, FALSE, NULL);
	if( _hTermEvent == NULL )
	{
		// If the function fails, the return value is NULL
		// Use GetLastError
		return E_UNEXPECTED;
	}


	return S_OK;
}

HRESULT PushToPullCaseProcessor::Reset()
{
	ReleaseCurrentCase();
	_currentCase = NULL;
	_lock.Term();
	_bTerminated		=	false;
	_hTerminationCode	=	S_OK;

	::CloseHandle( _hTermEvent );

	return _queueCache.Reset();
}

PushToPullCaseProcessor::~PushToPullCaseProcessor()
{
	Reset();
}

// IDMCaseProcessor

STDMETHODIMP	PushToPullCaseProcessor::RequestCapacity(ULONG in_cCapacity)
{
	// This function should not be called.

	return E_FAIL;
}

STDMETHODIMP	PushToPullCaseProcessor::ProcessCase(DM_CaseID dmcaseid, 
										   ULONG	cValue, 
										   ULONG*	rgValue)
{
	PullCASE*	pNewCase = NULL;
	DWORD		dwSize	=	0;

	HRESULT		hr	=	S_OK;
	hr	=	PullCASE::CreatePullCASE(_spidmmemoryallocator, &pNewCase);
	if( FAILED(hr) ) goto FINALIZE;

	// Compute the size to be occupied in cache by this case buffer
	dwSize		=	_bDensePushSet ? (cValue*sizeof(DM_STATE_VALUE)) : (cValue*sizeof(DM_ATTRIBUTE_VALUE));
	
	// Try to allocate a buffer of adequate size
	hr	=	_spidmmemoryallocator->Alloc( dwSize, (BYTE**)&pNewCase->rgCase );
	if( FAILED(hr) ) goto FINALIZE;

	pNewCase->caseID = dmcaseid;
	pNewCase->cAtts = cValue;
	pNewCase->dwSize = dwSize;

	memcpy( (void*)pNewCase->rgCase, rgValue, dwSize );


	// Wait for one of the following events to occur:
	//	- the queue has room to cache on more case	// WAIT_OBJECT_0
	//	- a termination condition occurred			// WAIT_OBJECT_0 + 1
	// Note that a successful termination (end of cases) cannot occur while 
	// a case is not processed (enqueued), so if the wait result is 'a termination condition occured) then
	// this is a genuine failure, regardless of the termination code
	HANDLE		arWaitObj[2];
	arWaitObj[0]	=	_queueCache.QueueHasRoomEvent();
	arWaitObj[1]	=	_hTermEvent;

	
	DWORD	dwWaitRet	=	::WaitForMultipleObjects( 2, (CONST HANDLE*)arWaitObj, FALSE, INFINITE);

	switch( dwWaitRet )
	{
		case WAIT_OBJECT_0:
			{
				// The queue has room to add a new case
				CComCritSecLock<CComCriticalSection>	autoLock(_lock);
				hr	=	_queueCache.AddCase( pNewCase );
				ATLASSERT( (S_OK == hr) || FAILED(hr) );
			}
			break;
		case (WAIT_OBJECT_0 + 1):
			{
				// Lock the access to the _hTerminationCode member
				CComCritSecLock<CComCriticalSection>	autoLock(_lock);

				// The termination event is signaled -- an error has occurred.
				// Return a failure code: either the termination code or E_UNEXPECTED
				hr	=	(SUCCEEDED(_hTerminationCode)) ? E_UNEXPECTED : _hTerminationCode;

			}
			break;
		default:
			// SHOULD NEVER BE HERE
			ATLASSERT(FALSE);
	}

FINALIZE:
	if( FAILED(hr) && pNewCase )
	{
		PullCASE::DeletePullCASE(_spidmmemoryallocator, pNewCase);
	}
	return hr;
}

STDMETHODIMP	PushToPullCaseProcessor::RestartCases()
{
	// Not implemented in this sample code.
	// Can be implemented as a message to the main thread that calls StartCases
	return E_NOTIMPL;
}
STDMETHODIMP	PushToPullCaseProcessor::IsDense()
{
	return _bDensePushSet ? S_OK : S_FALSE;
}

STDMETHODIMP	PushToPullCaseProcessor::GetNextCase(DM_CaseID* out_pdmcaseid, ULONG* out_pcValue, ULONG** out_prgValue)
{

	ReleaseCurrentCase();
	

	if( !out_pdmcaseid || !out_pcValue || !out_prgValue )
		return E_POINTER;

	HRESULT	hr	= E_UNEXPECTED;


	// Wait for one of the following events to occur:
	//	- the queue has data (at least one cached case)	// WAIT_OBJECT_0
	//	- a termination condition occurred				// WAIT_OBJECT_0 + 1
	// Note that a successful termination (end of cases) can occur while waiting
	// In this case, the queue might still have cached cases. In case of successful termination
	// this code checks whether the queue is empty. If it is not empty, it still fetches the case
	HANDLE		arWaitObj[2];
	arWaitObj[0]	=	_queueCache.QueueHasDataEvent();
	arWaitObj[1]	=	_hTermEvent;

	
	// WaitForMultipleObjects:  If more than one object became signalled during the call, this is the array index 
	// of the signalled object with the smallest index value of all the signalled objects
	// i.e. : if both events are signaled (the case set is completed and the queue has data) then the
	// returned code will be WAIT_OBJECT_0, i.e. the 'queue has data' event
	DWORD	dwWaitRet	=	::WaitForMultipleObjects( 2, (CONST HANDLE*)arWaitObj, FALSE, INFINITE);

	switch( dwWaitRet )
	{
		case WAIT_OBJECT_0:
			{
				// The queue has data. Mark the hr as S_OK, actual fetching will happen 
				// out of this switch
				hr	=	S_OK;
			}
			break;
		case (WAIT_OBJECT_0 + 1):
			{
				// Lock the access to the _hTerminationCode member
				CComCritSecLock<CComCriticalSection>	autoLock(_lock);
				
				// The termination event is signaled -- an error has occurred or the caseset is completed
				// If _hTerminationCode == S_OK, then the caseset is complete . Assert the queue is empty then return S_FALSE
				if( SUCCEEDED(_hTerminationCode) )
				{
					ATLASSERT( _queueCache.IsEmpty() );
					hr	=	S_FALSE;
				}
				else
				{
					hr	=	_hTerminationCode;
				}
			}
			break;
		default:
			// SHOULD NEVER BE HERE
			ATLASSERT(FALSE);
	}

	if( hr == S_OK )
	{	
		// The queue data event was signaled
		CComCritSecLock<CComCriticalSection>	autoLock(_lock);
		hr	=	_queueCache.RemoveCase(_currentCase);
		ATLASSERT( (hr == S_OK) || FAILED(hr) );

		if( S_OK == hr )
		{
			// Populate the output variables, if a case was fetched
			// Note that S_FALSE here is SUCCEEDED, but it means the cache is empty
			*out_pdmcaseid	=	_currentCase->caseID;
			*out_pcValue	=	_currentCase->cAtts;
			*out_prgValue	=	_currentCase->rgCase;
		}
	}

	
	return hr;
}

void	PushToPullCaseProcessor::ReleaseCurrentCase()
{
	if( _currentCase != NULL )
	{
		PullCASE::DeletePullCASE( _spidmmemoryallocator, _currentCase );
		_currentCase = NULL;
	}
}

HRESULT		PushToPullCaseProcessor::TerminateExecution(HRESULT hTerminationCode)
{
	// Receive the termination code from the StartCases thread
	CComCritSecLock<CComCriticalSection>	autoLock(_lock);
	if( !_bTerminated )
	{
		_bTerminated = true;
		_hTerminationCode	=	hTerminationCode;
		::SetEvent( _hTermEvent );
	}
	return S_OK;
}





PULLCASEPROCESSOR::PULLCASEPROCESSOR()
	: _vbCategorical(*this), _vdblValue(*this), _viAttribute(*this)
{
	_pcasereader = NULL;
	_bWorkingThreadTerminated = false;
	_hrWorkingThreadResult	=	S_OK;
}


HRESULT PULLCASEPROCESSOR::Initialize(IDMAttributeSet*	pidmattributeset, 
							      IDMPushCaseSet*	pidmpushcaseset, 
								  CASEREADER*		pcasereader,
								  bool				bMissingAtRandom) 
{
	// Prepare the case processor to receive the call to ProcessCases(). 

	ATLASSERT(pidmattributeset != NULL);
	ATLASSERT(pidmpushcaseset != NULL);
	ATLASSERT(pcasereader != NULL);



	if (pcasereader == NULL || pidmattributeset == NULL || pidmpushcaseset == NULL)
	{
		return E_POINTER;
	}

	_bWorkingThreadTerminated = false;
	_hrWorkingThreadResult	=	S_OK;


	_spidmpushcaseset = pidmpushcaseset;

	_pcasereader		= pcasereader;
	_cAttribute			= 0;
	_spidmattributeset	= pidmattributeset;
	_viewtype			= _pcasereader->Viewtype();
	_bDensePushSet		= (pidmpushcaseset->IsDense() == S_OK) ? true : false;
	_bMissingAtRandom	= bMissingAtRandom;

	HRESULT hr = pidmattributeset->GetAttributeCount(&_cAttribute);

	RETURN_ON_FAIL(hr);
    
	// We allow all four possibilities of:
	//
	// (source=dense, reader=dense)
	// (source=dense, reader=sparse)
	// (source=sparse, reader=dense)
	// (source=sparse, reader=sparse)

	// The member _vdblValue is passed to the reader: if the reader is dense, this
	// vector contains the value for attribute i at position i; if the reader is
	// sparse, the vector contains the value for attribute _viAttribute[i] at 
	// position i.

	if (_viewtype == CASEREADER::viewtypeDense)
	{ 
		CHECK_STL_MEM(_vdblValue.resize(_cAttribute));

		if (_bDensePushSet == false)
		{
			// Initialize with all missing

			for (ULONG iAttribute = 0; iAttribute < _cAttribute; iAttribute++)
			{
				_vdblValue[iAttribute] = dblMissing;
			}
		}
	}

	CHECK_STL_MEM(_vbCategorical.resize(_cAttribute));

	for (ULONG iAttribute = 0; iAttribute < _cAttribute; iAttribute++)
	{
		DM_ATTRIBUTE_FLAGS	dmattributeflags;

		hr = pidmattributeset->GetAttributeFlags(iAttribute, &dmattributeflags);

		RETURN_ON_FAIL(hr);

		if (dmattributeflags & DMAF_DISCRETE)
		{
			_vbCategorical[iAttribute] = true;
		}
		else
		{
			_vbCategorical[iAttribute] = false;
		}
	}

	return S_OK;
}

PULLCASEPROCESSOR::~PULLCASEPROCESSOR()
{
}






// Implementation of PULLCASEPROCESSOR
// This is exactly the same code as in CASEPROCESSOR::ProcessCase
HRESULT	PULLCASEPROCESSOR::ProcessCase(DM_CaseID dmcaseid, 
										   ULONG	cValue, 
										   ULONG*	rgValue)
{
	ULONG	ulCaseID = (UINT) dmcaseid;

	// The input argument rgValue is a bit confusing: if the push set is dense, 
	// then in_rgValue is an array of DM_STATE_VALUE. Otherwise, rgValue is an array
	// of DM_ATTRIBUTE_VALUE.

	// Notes on implementing 4 combinations : 
	//
	// (source=dense, reader=dense) : simply copy the (transformed) source
	// data values into _vdblValue.
	//
	// (source=dense, reader=sparse) : traverse the (transformed) source data
	// values, and for every value that is not missing (=dblMissing), push on
	// the value and attribute index to _vdblValue and _viAttribute, respectively.
	//
	// (source=sparse, reader=dense) : _vdblValue in this case is initially filled
	// with missing. For each source attribute/value, fill in the appropriate value
	// in _vdblValue. After calling CASEREADER::ProcessCaseDense(), traverse the
	// set of provided attributes again and fill with missing so that _vdblValue once
	// again contiains all missing.
	//
	// (source=sparse, reader=sparse) : simply copy the attribute indices and transformed
	// values to _viAttribute and _vdblValue, respectively.

	HRESULT hr = S_OK;

	if (_bDensePushSet)
	{
		if (_viewtype == CASEREADER::viewtypeSparse)
		{
			CHECK_STL_MEM(_vdblValue.resize(0));
			CHECK_STL_MEM(_viAttribute.resize(0));
		}

		if (cValue != _cAttribute)
		{
			// For dense data, we should get a value for every attribute.

			return E_UNEXPECTED;
		}

		DM_STATE_VALUE* rgdmstatevalue	= (DM_STATE_VALUE*) rgValue;
		
		// The values are specified in order. That is, rgdmstatevalue[i] is
		// the value for the ith attribute.

		for (UINT iAttribute = 0; iAttribute < _cAttribute; iAttribute++)
		{
			DBL dblValue;

			if (_vbCategorical[iAttribute])
			{
				dblValue = (_bMissingAtRandom) 
					? ::DblCategoricalMAR(rgdmstatevalue[iAttribute]) 
					: ::DblCategorical(rgdmstatevalue[iAttribute]);
			}
			else
			{
				dblValue = ::DblContinuous(rgdmstatevalue[iAttribute]);
			}

			if (_viewtype == CASEREADER::viewtypeSparse)
			{
				if (dblValue != dblMissing)
				{
					CHECK_STL_MEM(_vdblValue.push_back(dblValue));
					CHECK_STL_MEM(_viAttribute.push_back(iAttribute));
				}
			}
			else
			{
				_vdblValue[iAttribute] = dblValue;
			}
		}

		if (_viewtype == CASEREADER::viewtypeSparse)
		{
			hr = _pcasereader->ProcessCaseSparse(ulCaseID, _viAttribute, _vdblValue);

			return hr;
		}
		else
		{
			hr = _pcasereader->ProcessCaseDense(ulCaseID, _vdblValue);

			return hr;
		}
	}
	else
	{
		// Sparse input data 

		DM_ATTRIBUTE_VALUE*	rgdmattributevalue = (DM_ATTRIBUTE_VALUE*) rgValue;

		if (_viewtype == CASEREADER::viewtypeSparse)
		{
			CHECK_STL_MEM(_vdblValue.resize(cValue));
		}

		// Always cache away the attribute indices. If we have a dense reader, we 
		// need this list to re-initialize to missing.

		CHECK_STL_MEM(_viAttribute.resize(cValue));

		for (UINT idmattributevalue = 0; idmattributevalue < cValue; idmattributevalue++)
		{
			UINT			iAttribute		= rgdmattributevalue[idmattributevalue].Attribute;
			DM_STATE_VALUE	dmstatevalue	= rgdmattributevalue[idmattributevalue].Value;
			DBL				dblValue;

			if (_vbCategorical[iAttribute])
			{
				dblValue = (_bMissingAtRandom) 
					? ::DblCategoricalMAR(dmstatevalue) 
					: ::DblCategorical(dmstatevalue);
			}
			else
			{
				dblValue = ::DblContinuous(dmstatevalue);
			}

			if (_viewtype == CASEREADER::viewtypeSparse)
			{
				_viAttribute[idmattributevalue]	= iAttribute;
				_vdblValue[idmattributevalue]	= dblValue;
			}
			else
			{
				ATLASSERT(_vdblValue[iAttribute] == dblMissing);

				_vdblValue[iAttribute] = dblValue;
			}
		}

		if (_viewtype == CASEREADER::viewtypeSparse)
		{
			hr = _pcasereader->ProcessCaseSparse(ulCaseID, _viAttribute, _vdblValue);

			return hr;
		}
		else
		{
			hr = _pcasereader->ProcessCaseDense(ulCaseID, _vdblValue);

			// Reinitialize dense values to missing

			for (UINT iValue = 0; iValue < cValue; iValue++)
			{
				_vdblValue[_viAttribute[iValue]] = dblMissing;
			}

			return hr;
		}
	}
}




HRESULT	PULLCASEPROCESSOR::ProcessCases(IDMContextServices *in_pExeContext)
{
	ATLASSERT( _spidmpushcaseset );
	DMHALLOC	dmhalloc;			// Memory-allocation wrapper
	HRESULT hr	=	dmhalloc.SetAllocator( in_pExeContext );
	RETURN_ON_FAIL(hr);


	// Create a PushToPullCaseProcessor
	CComObjectDM<PushToPullCaseProcessor>*	pcaseprocessor = NULL;
	hr = CComObjectDM<PushToPullCaseProcessor>::CreateInstance(dmhalloc, &pcaseprocessor);
	RETURN_ON_FAIL(hr);

	// Initialize the PushToPullCaseProcessor
	hr	=	pcaseprocessor->Initialize( _spidmpushcaseset, PULL_CASESET_CACHE_SIZE );
	RETURN_ON_FAIL(hr);

	// QI the PushToPullCaseProcessor for IDMCaseProcessor
	hr = pcaseprocessor->QueryInterface(&_spidmcaseprocessor);
	RETURN_ON_FAIL(hr);

	// QI the PushToPullCaseProcessor for IDMPushToPullCaseSetHelper
	hr = pcaseprocessor->QueryInterface(&_spidmpullcaseset);
	RETURN_ON_FAIL(hr);

	// Create a new thread to process the cases in a 'pull' manner
	// 
	// The new thread should inherit the security descriptor of this thread
	// If lpThreadAttributes is NULL, the thread gets a default security descriptor. 
	// The ACLs in the default security descriptor for a thread come from the primary or impersonation token of the creator.
	// 
	HANDLE hNewThread = ::CreateThread( NULL, 0, CaseProcessingProc, (LPVOID)this, 0, NULL);
	if( hNewThread == NULL )
	{
		return E_FAIL;	// error in creating the new thread
	}

	
	
	// Start the cases
	hr	=	_spidmpushcaseset->StartCases(_spidmcaseprocessor, false /*need case id*/);



	// Signal termination of cases to the PushToPull case set
	_spidmpullcaseset->TerminateExecution( hr );

	// - if the working thead is not finalized
	//		Wait for the working thread to finish
	// Return the result of the working thread (with some sanity checks)

	if( !_bWorkingThreadTerminated )
	{
		::WaitForSingleObject(hNewThread, INFINITE);
	}

	if( FAILED(hr) )
	{
		// If StartCases failed
		//		If the working thread failed, then 
		//		its' return code will contain the most accurrate error code
		if( FAILED(_hrWorkingThreadResult) )
			hr	=	_hrWorkingThreadResult;
		// else, keep the failed hr from StartCases
	}
	else
	{
		// StartCases ended successfully, return whatever comes from the working thread
		hr	=	_hrWorkingThreadResult;
	}

	_spidmcaseprocessor = NULL;
	_spidmpullcaseset	= NULL;


	return hr;
}


DWORD WINAPI	PULLCASEPROCESSOR::CaseProcessingProc(LPVOID lpParam)
{
	PULLCASEPROCESSOR*	pThis	=	(PULLCASEPROCESSOR*)lpParam;
	CComPtr<IDMPushToPullCaseSetHelper>	spPullCaseSet;
	spPullCaseSet = pThis->_spidmpullcaseset;
	ATLASSERT( spPullCaseSet != NULL );
	


	
	// Start the cases on this thread
	HRESULT		hr	=	S_OK;
	
	// Now, the secondary thread is populating the cache
	// 'Pull' the cases from cache here
	while( TRUE )
	{
		DM_CaseID	caseID	=	0;
		ULONG		cValues	=	0;
		ULONG*		rgValues=	NULL;

		// Get the first case in cache
		hr	=	spPullCaseSet->GetNextCase( &caseID, &cValues, &rgValues);

		if( SUCCEEDED(hr) && hr != S_OK )
		{
			// Case completed, cache is empty and the worker thread is completed
			// Terminate and return S_OK
			hr	=	S_OK;
			break;
		}
		else if( FAILED(hr) )
		{
			// error in fetching the next case from the pull case processor.
			// Terminate and return this error
			spPullCaseSet->TerminateExecution( hr );
			break;
		}
		else 
		{
			ATLASSERT( hr == S_OK  );
			hr	=	pThis->ProcessCase( caseID, cValues, rgValues);
			if( FAILED(hr) )
			{
				// Terminate and return this error
				spPullCaseSet->TerminateExecution( hr );
				break;
			}
		}
	}
	pThis->SetCaseThreadExitCode(hr);

	return 0;
}

void	PULLCASEPROCESSOR::SetCaseThreadExitCode(HRESULT hRet)
{
	if( !_bWorkingThreadTerminated )
	{
		_bWorkingThreadTerminated	=	true;
		_hrWorkingThreadResult		=	hRet;
	}
}

